package com.task.scheduler.queue;

import com.task.scheduler.config.TaskSchedulerProperties;
import com.task.scheduler.enums.TaskStatus;
import com.task.scheduler.event.TaskEvent;
import com.task.scheduler.executor.TaskExecutor;
import com.task.scheduler.executor.TaskType;
import com.task.scheduler.model.GenericTask;
import com.task.scheduler.persistence.entity.TaskEntity;
import com.task.scheduler.persistence.entity.TaskExecutionEntity;
import com.task.scheduler.persistence.repository.TaskExecutionRepository;
import com.task.scheduler.persistence.repository.TaskRepository;
import com.task.scheduler.service.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Service
@Slf4j
public class RedissonTaskQueueService implements TaskService {

    private final RedissonClient redissonClient;
    private final List<TaskExecutor> taskExecutors;
    private final ApplicationEventPublisher eventPublisher;
    private final Map<String, TaskExecutor> executorMap = new HashMap<>();

    @Autowired
    private TaskSchedulerProperties taskSchedulerProperties; // ✅ 修改为统一配置类
    @Autowired
    private TaskExecutionRepository taskExecutionRepository;
    @Autowired
    private TaskRepository taskRepository;


    public RedissonTaskQueueService(RedissonClient redissonClient,
                                    List<TaskExecutor> taskExecutors,
                                    ApplicationEventPublisher publisher) {
        this.redissonClient = redissonClient;
        this.taskExecutors = taskExecutors;
        this.eventPublisher = publisher;

        Map<String, TaskExecutor> executorMap = new HashMap<>();

        for (TaskExecutor executor : taskExecutors) {
            Class<? extends TaskExecutor> clazz = executor.getClass();
            if (clazz.isAnnotationPresent(TaskType.class)) {
                TaskType annotation = clazz.getAnnotation(TaskType.class);
                for (String type : annotation.value()) {
                    executorMap.put(type, executor);
                }
            }
        }

        // ✅ 把局部变量赋值给类成员变量
        this.executorMap.putAll(executorMap);

        if (executorMap.isEmpty()) {
            log.warn("⚠️ 没有找到任何 TaskExecutor");
        } else {
            log.info("✅ 共找到 {} 个 TaskExecutor", executorMap.size());
            for (String taskType : executorMap.keySet()) {
                log.info("🧵 启动消费者线程，监听任务类型：{}", taskType);
                startConsumingInBackground(taskType);
            }
        }
        log.info("✅ 注入的执行器列表：" + taskExecutors);
    }

    @Override
    public void submit(GenericTask task) {
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue("task:queue:" + task.getTaskType());
        RDelayedQueue<String> delayedQueue = redissonClient.getDelayedQueue(queue);

        String taskJson = task.toJson();
        delayedQueue.offer(taskJson, task.getDelaySeconds(), TimeUnit.SECONDS);
        log.info("任务 [{}] 已入队，类型: {}, 延迟: {}s", task.getTaskId(), task.getTaskType(), task.getDelaySeconds());
    }

    @Override
    public void cancel(String taskId) {
        // 遍历所有任务类型队列，尝试移除任务
        for (String taskType : executorMap.keySet()) {
            RBlockingQueue<String> queue = redissonClient.getBlockingQueue("task:queue:" + taskType);
            queue.remove(taskId);  // 从队列中移除任务
            log.info("任务 [{}] 已从队列 [{}] 中移除", taskId, taskType);
        }
    }

    /**
     * 启动后台消费线程，监听指定任务类型的队列
     */
    public void startConsumingInBackground(String taskType) {
        ExecutorService scheduler = Executors.newSingleThreadExecutor();
        scheduler.submit(() -> startConsuming(taskType));
    }

    /**
     * 消费指定任务类型的消息
     */
    public void startConsuming(String taskType) {
        RBlockingQueue<String> queue = redissonClient.getBlockingQueue("task:queue:" + taskType);
        while (!Thread.currentThread().isInterrupted()) {
            try {
                String taskJson = queue.take(); // 阻塞等待任务
                GenericTask task = GenericTask.fromJson(taskJson);

                // 必须使用 SDK 自动注册的执行器
                TaskExecutor matchedExecutor = executorMap.get(task.getTaskType());

                // 没有匹配的执行器，跳过执行
                if (matchedExecutor == null) {
                    log.warn("⚠️ 未找到匹配执行器，请检查 taskType 或注册自定义 TaskExecutor");
                    return;
                }

                String result = null;
                TaskStatus status = TaskStatus.FAILED;

                try {
                    result = matchedExecutor.execute(task);
                    status = TaskStatus.SUCCESS;
                } catch (Exception e) {
                    task.setLastErrorMessage(e.getMessage());  // 记录失败原因
                    log.error("任务 [{}] 执行失败: {}", task.getTaskId(), e.getMessage(), e);  // 输出完整堆栈
                    if (task.getRetryCount() < task.getMaxRetry()) {
                        task.setRetryCount(task.getRetryCount() + 1);
                        submit(task); // 重新入队
                    }
                }

                eventPublisher.publishEvent(new TaskEvent(this, task.getTaskType(), task, result, status));


                // ✅ 持久化逻辑开始
                if (taskSchedulerProperties.getPersistence().isEnabled()) {
                    TaskEntity taskEntity = new TaskEntity();
                    taskEntity.setId(task.getTaskId());
                    taskEntity.setTaskType(task.getTaskType());
                    taskEntity.setJsonParams(task.getJsonParams());
                    taskEntity.setMaxRetry(task.getMaxRetry());
                    taskEntity.setDelaySeconds(task.getDelaySeconds());
                    taskEntity.setExecutionMode(task.getExecutionMode());

                    TaskExecutionEntity executionEntity = new TaskExecutionEntity();
                    executionEntity.setTaskId(task.getTaskId());
                    executionEntity.setRetryCount(task.getRetryCount());
                    executionEntity.setStatus(status);
                    executionEntity.setResult(result);
                    executionEntity.setErrorMessage(task.getLastErrorMessage());

                    taskRepository.save(taskEntity);
                    taskExecutionRepository.save(executionEntity);

                    // 更新 latest_execution_id
                    taskEntity.setLatestExecutionId(executionEntity.getId());
                    taskRepository.save(taskEntity);
                }
                // ✅ 持久化逻辑结束

            } catch (Exception e) {
                log.error("消费任务时发生异常", e);
            }
        }
    }
}
